package com.example.sourcecodelearning.concurrent.blockingqueue;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author yuzhenxin
 * @description 模拟实现 ArrayBlockingQueue
 * @date 2023-07-15 15:43
 */
@SuppressWarnings("unchecked")
public class CustomBlockingQueue<E> {
    private final ReentrantLock lock;
    private final Condition notEmpty;
    private final Condition notFull;
    private int takeIndex;
    private int putIndex;
    private int count;
    private final E[] queue;

    public CustomBlockingQueue(int size) {
        if (size <= 0) {
            throw new IllegalArgumentException();
        }
        this.lock = new ReentrantLock();
        this.notFull = lock.newCondition();
        this.notEmpty = lock.newCondition();
        this.queue = (E[]) new Object[size];
    }

    /**
     * 阻塞提交
     *
     * @param element 待添加的元素
     */
    public void put(E element) {
        lock.lock();
        try {
            while (count == queue.length) {
                notFull.await();
            }
            enqueue(element);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 非阻塞提交
     *
     * @param element 待添加的元素
     * @return 是否提交成功
     */
    public boolean offer(E element) {
        lock.unlock();
        try {
            if (count == queue.length) {
                // 数组满了直接返回
                return false;
            }
            enqueue(element);
            return true;
        } finally {
            lock.unlock();
        }
    }

    public boolean add(E element) {
        if (offer(element)) {
            return true;
        } else {
            throw new RuntimeException("入队失败");
        }
    }

    private void enqueue(E element) {
        this.queue[putIndex] = element;
        putIndex = (putIndex + 1) % queue.length;
        count++;
        notEmpty.signal();
    }

    /**
     * 阻塞获取
     *
     * @return 队头元素
     * @throws InterruptedException if interrupted while waiting
     */
    public E take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0) {
                notEmpty.await();
            }
            return dequque();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 非阻塞获取
     *
     * @return 队头元素
     */
    public E poll() {
        lock.lock();
        try {
            if (count == 0) {
                return null;
            }
            return dequque();
        } finally {
            lock.unlock();
        }
    }

    private E dequque() {
        E e = this.queue[takeIndex];
        // help gc
        queue[takeIndex] = null;
        takeIndex = (takeIndex + 1) % queue.length;
        count--;
        notFull.signal();
        return e;
    }

    @Override
    public String toString() {
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append("[");
        // 遍历数组的时候不能进行插入和删除
        lock.lock();
        try {
            if (count == 0) {
                stringBuilder.append("]");
            } else {
                int cur = 0;
                while (cur != count) {
                    stringBuilder.append(queue[(takeIndex + cur) % queue.length])
                            .append(",");
                    cur++;
                }
                stringBuilder.delete(stringBuilder.length() - 2, stringBuilder.length())
                        .append("]");
            }
        } finally {
            lock.unlock();
        }
        return stringBuilder.toString();
    }
}
